package com.shujia.trip.ods;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;

public class OdsKafkaActivity {
    public static void main(String[] args) {
        EnvironmentSettings settings = EnvironmentSettings
                .newInstance()
                // 指定为流处理的运行模式
                .inStreamingMode()
                // 指定为批处理的运行模式
                //.inBatchMode()
                .build();

        // 从Kafka消费student数据，统计班级人数

        // 第二种方式构建Table环境
        TableEnvironment tEnv = TableEnvironment.create(settings);

        tEnv.executeSql("CREATE TABLE ods_cdc_mysql_activity (\n" +
                "  activity_id BIGINT comment '主键'\n" +
                "  ,cid STRING comment '企业ID'\n" +
                "  ,name STRING comment '活动名称'\n" +
                "  ,`type` STRING comment '活动类型 BUBBLE：冒泡'\n" +
                "  ,start_time TIMESTAMP(3) comment '开始时间'\n" +
                "  ,end_time TIMESTAMP(3) comment '结束时间'\n" +
                "  ,is_default INT comment ''\n" +
                "  ,`status` INT comment '活动状态 0-草稿、1-待启用'\n" +
                "  ,is_delete INT comment '是否删除 0-未删除、1-删除'\n" +
                "  ,creator_id STRING comment '创建者id'\n" +
                "  ,nick_name STRING comment '创建者名称'\n" +
                "  ,update_user_id STRING comment '最后修改人id'\n" +
                "  ,update_nick_name STRING comment '最后修改人名称'\n" +
                "  ,gmt_create TIMESTAMP(3) comment '创建时间'\n" +
                "  ,gmt_modified TIMESTAMP(3) comment '更新时间'\n" +
                "  ,PRIMARY KEY (activity_id) NOT ENFORCED\n" +
                ") WITH (\n" +
                "    'connector' = 'mysql-cdc',\n" +
                "    'hostname' = 'master',\n" +
                "    'port' = '3306',\n" +
                "    'username' = 'root',\n" +
                "    'password' = '123456',\n" +
                "    'database-name' = 'trip',\n" +
                "    'table-name' = 'activity'\n" +
                ");");

        tEnv.executeSql("CREATE TABLE ods_kafka_activity (\n" +
                "  activity_id BIGINT comment '主键'\n" +
                "  ,cid STRING comment '企业ID'\n" +
                "  ,name STRING comment '活动名称'\n" +
                "  ,`type` STRING comment '活动类型 BUBBLE：冒泡'\n" +
                "  ,start_time TIMESTAMP(3) comment '开始时间'\n" +
                "  ,end_time TIMESTAMP(3) comment '结束时间'\n" +
                "  ,is_default INT comment ''\n" +
                "  ,`status` INT comment '活动状态 0-草稿、1-待启用'\n" +
                "  ,is_delete INT comment '是否删除 0-未删除、1-删除'\n" +
                "  ,creator_id STRING comment '创建者id'\n" +
                "  ,nick_name STRING comment '创建者名称'\n" +
                "  ,update_user_id STRING comment '最后修改人id'\n" +
                "  ,update_nick_name STRING comment '最后修改人名称'\n" +
                "  ,gmt_create TIMESTAMP(3) comment '创建时间'\n" +
                "  ,gmt_modified TIMESTAMP(3) comment '更新时间'\n" +
                "  ,PRIMARY KEY (activity_id) NOT ENFORCED\n" +
                ") WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = 'ods_activity', -- 数据的topic\n" +
                "  'properties.bootstrap.servers' = 'master:9092,node1:9092,node2:9092', -- broker 列表\n" +
                "  'properties.group.id' = 'trip', -- 消费者组\n" +
                "  'scan.startup.mode' = 'earliest-offset', -- 读取数据的位置earliest-offset latest-offset\n" +
                "  'format' = 'canal-json' -- 保存更新的流\n" +
                ");");

        tEnv.executeSql("insert into ods_kafka_activity\n" +
                "select * from ods_cdc_mysql_activity;");

    }
}
